package com.yyf.kafka.config;

import com.yyf.kafka.dto.MessageEntity;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * 壹 * @Author: yang
 * 貮 * @Date: 2019/11/5 11:35
 */

@Configuration
@EnableKafka
public class KafkaProducerConfig {

	@Value("${kafka.producer.servers}")
	private String servers;

	@Value("${kafka.producer.retries}")
	private String retries;

	@Value("${kafka.producer.batch.size}")
	private int batchSize;

	@Value("${kafka.producer.linger}")
	private int linger;

	@Value("${kafka.producer.buffer.memory}")
	private int bufferMemory;


	public Map<String,Object> producerConfigs(){
		Map<String,Object> props = new HashMap<>();
		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,servers);
		props.put(ProducerConfig.RETRIES_CONFIG,retries);
		props.put(ProducerConfig.BATCH_SIZE_CONFIG,batchSize);
		props.put(ProducerConfig.LINGER_MS_CONFIG,linger);
		props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,bufferMemory);
		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
		return props;
	}

	public ProducerFactory<String,MessageEntity> producerFactory(){
		return new DefaultKafkaProducerFactory<>(
				producerConfigs(),
				new StringSerializer(),
				new JsonSerializer<MessageEntity>()
		);
	}

	@Bean
	public KafkaTemplate<String,MessageEntity> kafkaTemplate(){
		return new KafkaTemplate<>(producerFactory());
	}
}
